package com.j.lemon.learn.rocketmq;

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;

public class Consumer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demoConsumer1",
//                new AclClientRPCHook(new SessionCredentials("rocketmq2","12345678")),
//                new AllocateMessageQueueAveragely());
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demoConsumer1");

//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.setNamesrvAddr("10.28.149.182:9876;10.28.149.188:9876;10.28.149.197:9876");
        consumer.setInstanceName("QuickStartConsumer");
        consumer.subscribe("topic2", "t1");

//
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            System.out.println(Thread.currentThread().getName() + " 接收新消息: " + msgs);
            for (Message msg :msgs){
                System.out.println("消息内容："+new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

//        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
//            for (MessageExt msg:list){
//                System.out.println(msg.getMsgId()+"---"+new String(msg.getBody()));
//            }
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
        consumer.start();
        System.out.println("消费者已启动");
        Thread.sleep(100000);
    }

}
